package com.example.service;

import com.example.annotation.RabbitmqListener;
import com.example.annotation.RabbitmqProp;
import com.example.listener.MQMessageListener;
import com.example.utils.IdUtils;
import com.rabbitmq.client.*;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author chentao
 * @version 1.0
 * @description: RabbitMQ的帮助类
 * 死信队列 x-dead-letter-routing-key 死信路由key
 * @date 2025/3/18 下午9:44
 */
@Data
public class SysRabbitMQServiceImpl implements SysRabbitMQService, BeanPostProcessor, ApplicationContextAware, InitializingBean, ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Map<String, RabbitmqListener> beanListenerMaps = new HashMap<String, RabbitmqListener>();

    private ConnectionFactory connectionFactory;

    private Connection connection;

    private Channel channel;

    private String host;

    private int port;

    private String username;

    private String password;

    private ApplicationContext applicationContext;

    private ExecutorService executorService;

    @Override
    public void destroy() throws Exception {
        this.connection.close();
        this.executorService.shutdownNow();
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean.getClass().isAnnotationPresent(RabbitmqListener.class) && bean instanceof MQMessageListener) {
            RabbitmqListener rabbitmqListener = bean.getClass().getAnnotation(RabbitmqListener.class);
            this.beanListenerMaps.put(beanName, rabbitmqListener);
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.startClient();
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() == null) {
            for (String beanName : beanListenerMaps.keySet()) {
                this.startListening(beanName);
            }
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public Channel getChannel() {
        return this.channel;
    }

    @Override
    public void startClient() throws Exception {
        this.executorService = Executors.newCachedThreadPool();
        this.connectionFactory = new ConnectionFactory();
        this.connectionFactory.setHost(host);
        this.connectionFactory.setPort(port);
        this.connectionFactory.setUsername(username);
        this.connectionFactory.setPassword(password);
        this.connectionFactory.setAutomaticRecoveryEnabled(true);
        this.connection = this.connectionFactory.newConnection();
        this.channel = this.connection.createChannel();
    }

    @Override
    public void startListening(String beanName) {
        this.executorService.execute(new Runnable() {
            @Override
            public void run() {
                Channel channel = null;
                Map<String, Object> map = new HashMap<String, Object>();
                RabbitmqListener rabbitmqListener = beanListenerMaps.get(beanName);
                String queueName = rabbitmqListener.queues();
                if (rabbitmqListener.type() == BuiltinExchangeType.FANOUT) {
                    queueName = queueName + IdUtils.getId();
                }
                for (RabbitmqProp prop : rabbitmqListener.props()) {
                    if (StringUtils.hasText(prop.value())) {
                        map.put(prop.key(), prop.value());
                    } else {
                        map.put(prop.key(), prop.boolValue());
                    }
                }
                try {
                    channel = connection.createChannel();
                    channel.exchangeDeclare(rabbitmqListener.exchange(), rabbitmqListener.type());
                    channel.queueDeclare(queueName, rabbitmqListener.durable(), rabbitmqListener.exclusive(), rabbitmqListener.autoDelete(), map);
                    channel.queueBind(queueName, rabbitmqListener.exchange(), rabbitmqListener.routingKey());
                    channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                            MQMessageListener mqMessageListener = null;
                            try {
                                mqMessageListener = (MQMessageListener) applicationContext.getBean(beanName);
                                mqMessageListener.onMessage(mqMessageListener, consumerTag, envelope, properties, body);
                            } catch (Exception e) {
                                logger.warn("queue 消费消息失败: " + e.getMessage());
                            } finally {
                                logger.info("记录mq监听消息的日志 ...");
                            }
                        }
                    });
                    logger.warn("exchange: " + rabbitmqListener.exchange() + ", type: " + rabbitmqListener.type() + ", routingKey: " + rabbitmqListener.routingKey() + ", queueName: " + queueName + " 开启消息的监听 ...");
                } catch (Exception e) {
                    if (channel != null) {
                        try {
                            channel.close();
                        } catch (Exception ee) {
                            logger.warn("channel close error : " + ee.getMessage());
                        }
                    }
                    logger.error("rabbitmq listener start error: " + "exchange: " + rabbitmqListener.exchange() + ", type: " + rabbitmqListener.type() + ", routingKey: " + rabbitmqListener.routingKey() + ", queueName: " + queueName + " 监听失败 ...");
                }
            }
        });
    }
}
